package cn.lsh.main;

import cn.lsh.comparator.LineSortComparator;
import cn.lsh.mapper.WeatherSortMapper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class SortDriver extends BaseDriver {

	public static void main(String[] args) throws Exception {
		localClusterRun(args);
	}

	/**
	 * windows本地运行
	 *
	 * @param args
	 * @throws Exception
	 */
	public static void localRun(String[] args) throws Exception {
		buildJob(true);
		//IDE本地运行
		conf.set("mapreduce.framework.name", "local");

		job.setMapperClass(WeatherSortMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		job.setSortComparatorClass(LineSortComparator.class);

		FileInputFormat.addInputPath(job, new Path("file:/bigdata/hadoop-test/input/ncdc"));
		FileOutputFormat.setOutputPath(job, new Path("file:/bigdata/hadoop-test/output/ncdc"));

		job.waitForCompletion(true);
	}

	/**
	 * windows提交，yarn集群运行
	 *
	 * @param args
	 * @throws Exception
	 */
	public static void localClusterRun(String[] args) throws Exception {
		Configuration conf = new Configuration(true);
		conf.set("mapreduce.framework.name", "yarn");
		conf.set("mapreduce.app-submission.cross-platform", "true");
		conf.set("yarn.resourcemanager.cluster-id", "cluster1");
		//通过这种方式设置java客户端访问hdfs的身份
		System.setProperty("HADOOP_USER_NAME", "root");

		Job job = Job.getInstance(conf);
		job.setJobName("local-cluster");
		job.setJarByClass(SortDriver.class);
		job.setJar("G:\\bigdata\\hadoop-test\\target\\hadoop-test-1.0-SNAPSHOT.jar");

		job.setMapperClass(WeatherSortMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		job.setSortComparatorClass(LineSortComparator.class);

		FileInputFormat.addInputPath(job, new Path("/test/input/ncdc"));
		FileOutputFormat.setOutputPath(job, new Path("/test/output/ncdc"));

		job.waitForCompletion(true);
	}

}
